package com.bingxu.flink.state;

import com.bingxu.flink.bean.InputNumber;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author :BingXu
 * @description :TODO 连续输入数字大于100
 * @date :2021/8/19 14:58
 * @modifier :
 */

public class KeyValueState {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 10086);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);

        DataStreamSource<String> sourceStream = env.socketTextStream("localhost", 9999);

        sourceStream
                .map(new MapFunction<String, InputNumber>() {
                    @Override
                    public InputNumber map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new InputNumber(fields[0], Integer.parseInt(fields[1]));

                    }
                })
                .keyBy(ele -> ele.getId())
                .process(new KeyedProcessFunction<String, InputNumber, String>() {

                    private ValueState<Integer> valueState;
                    private int init = 0;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        valueState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("cnt", Integer.class));
                    }

                    @Override
                    public void processElement(InputNumber value, Context ctx, Collector<String> out) throws Exception {
//                        if (valueState.value() == null && value.getNum() > 100) {
//                            init++;
//                            valueState.update(init);
//                        } else if(valueState.value()!=null) {
//                            if (value.getNum() > 100) {
//                                if (valueState.value() >= 1) {
//                                    out.collect("连续" + valueState.value() + "次大于100，请注意。。。。");
//
//                                } else {
//                                    init++;
//                                    valueState.update(init);
//                                }
//                            }else{
//                                init = 0;
//                                valueState.update(init);
//                            }
//
//                        }
                        if (value.getNum() > 99) {
                            if (valueState.value() == null || valueState.value() < 1) {
                                init++;
                                valueState.update(init);
                            } else {
                                if (valueState.value() >= 1) {
                                    init++;
                                    out.collect("当前用户连续" + (valueState.value() + 1) + "次大于99，关注关注。。。。");
                                    valueState.update(init);
                                }
                            }
                        } else {
                            init=0;
                            valueState.update(init);
                        }

                    }
                }).print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
